package com.imooc.producer.a_normal;

import com.utils.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/9-2024
 * @Description: 消费者消费消息。
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        //定义消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer");
        consumer.setNamesrvAddr(Const.NAME_SERVER_SINGLE_LOCAL);

        //定义消费模式为 ‘集群消费’（默认的就是集群消费模式）
        //只有集群消费模式下才有 【reblance】 重平衡机制。
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //消费者订阅主题
        consumer.subscribe("sync_topic","*");
        consumer.subscribe("async_topic","*");
        consumer.subscribe("oneway_topic","*");
        //消费者监听器(处理消息的逻辑)
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (int i = 0; i < msgs.size(); i++) {
                    System.out.println(new String(msgs.get(i).getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //启动消费者
        consumer.start();
    }
}